 /***********************************************************
 *  File_Name  : XJThreadManager.cpp
 *  File_Path   : 
 *  File_Encode : UTF8
 *  Description : 线程管理类
 *  
 *  OS : Linux, UNIX
 *  LastVersion  : 20130723
 *      
 *  Author      : LiXiujie(李秀杰)
 *  Date        : 2013-7-23
 *  Version     : 20130723
 *  Description : 
 *                
 *  Edit Author : 
 *  Edit Date   : 
 *  Version     : 
 *  Description :
 *                
 ***********************************************************/

#include "XJThreadManager.h"
#include <unistd.h>
#include <string.h>

namespace XJUtilClass {

// std::vector<XJThreadManager::XJThreadInfo *> XJThreadManager::m_veThreadInfos;
XJThreadManager * 	XJThreadManager::m_pThis = NULL;
pthread_mutex_t XJThreadManager::m_ThreadMutex;

XJThreadManager::XJThreadManager() {

}

XJThreadManager::~XJThreadManager() {
}

/*方法：单实例
 * */
XJThreadManager *XJThreadManager::instance (){
	if (m_pThis == NULL){
		m_pThis = new XJThreadManager();
		pthread_mutex_init(&m_ThreadMutex, NULL);
	}
	return m_pThis;
}

/*方法：单实例
 * */
void XJThreadManager::close_singleton (){
	if (m_pThis){
		m_pThis->kill_all();
		m_pThis->wait_all(0);
		delete m_pThis;
		m_pThis = NULL;
		pthread_mutex_destroy(&m_ThreadMutex);
	}
}
/*方法：创建一个新线程
 * */
int XJThreadManager::spawn(XJ_THR_FUNC func,
		void *arg,
		long flags,
		pthread_t *t_id,
		long priority,
		int grp_id,
		size_t stack_size){
	return spawn_i(1, func, arg,flags, t_id, priority, grp_id, stack_size);
}
int XJThreadManager::spawn_n(size_t n,
		XJ_THR_FUNC func,
		void *arg,
		long flags,
		pthread_t *t_id,
		long priority,
		int grp_id,
		size_t stack_size){
	return spawn_i(n, func, arg,flags, t_id, priority, grp_id, stack_size);
}
/*方法：终止一个线程
 * */
int XJThreadManager::XJThreadManager::exit_all(){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->exit == 1 ||
				m_veThreadInfos[i]->t_id == pthread_self()
		){
			m_veThreadInfos[i]->opt = XJ_THR_O_EXIT;
			m_veThreadInfos[i]->t_opt = pthread_self();
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::XJThreadManager::exit(pthread_t t_id){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == t_id){
			if (m_veThreadInfos[i]->exit == 1 ||
					m_veThreadInfos[i]->t_id == pthread_self()
			){
				m_veThreadInfos[i]->opt = XJ_THR_O_EXIT;
				m_veThreadInfos[i]->t_opt = pthread_self();
			}
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::XJThreadManager::exit_group(int group_id){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->group == group_id){
			if (m_veThreadInfos[i]->exit == 1 ||
					m_veThreadInfos[i]->t_id == pthread_self()
			){
				m_veThreadInfos[i]->opt = XJ_THR_O_EXIT;
				m_veThreadInfos[i]->t_opt = pthread_self();
			}
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::XJThreadManager::exit_fun(XJ_THR_FUNC func){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->fun == func){
			if (m_veThreadInfos[i]->exit == 1 ||
					m_veThreadInfos[i]->t_id == pthread_self()
			){
				m_veThreadInfos[i]->opt = XJ_THR_O_EXIT;
				m_veThreadInfos[i]->t_opt = pthread_self();
			}
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::XJThreadManager::exit_self(){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == pthread_self()){
			m_veThreadInfos[i]->opt = XJ_THR_O_EXIT;
			m_veThreadInfos[i]->t_opt = pthread_self();
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}

/*方法：线程管理器暂停指定的线程
 * */
int XJThreadManager::suspend_all(){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		m_veThreadInfos[i]->opt = XJ_THR_O_SUSPEND;
		m_veThreadInfos[i]->t_opt = pthread_self();
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::suspend(pthread_t t_id){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == t_id){
			m_veThreadInfos[i]->opt = XJ_THR_O_SUSPEND;
			m_veThreadInfos[i]->t_opt = pthread_self();
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::suspend_group(int group_id){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->group == group_id){
			m_veThreadInfos[i]->opt = XJ_THR_O_SUSPEND;
			m_veThreadInfos[i]->t_opt = pthread_self();
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::suspend_fun(XJ_THR_FUNC func){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->fun == func){
			m_veThreadInfos[i]->opt = XJ_THR_O_SUSPEND;
			m_veThreadInfos[i]->t_opt = pthread_self();
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::suspend_self(){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == pthread_self()){
			m_veThreadInfos[i]->opt = XJ_THR_O_SUSPEND;
			m_veThreadInfos[i]->t_opt = pthread_self();
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
/*方法：线程管理器恢复执行前面暂停的线程
 * */
int XJThreadManager::resume_all(){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		m_veThreadInfos[i]->opt = XJ_THR_O_RESUME;
		m_veThreadInfos[i]->t_opt = pthread_self();
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::resume(pthread_t t_id){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == t_id){
			m_veThreadInfos[i]->opt = XJ_THR_O_RESUME;
			m_veThreadInfos[i]->t_opt = pthread_self();
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::resume_group(int group_id){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->group == group_id){
			m_veThreadInfos[i]->opt = XJ_THR_O_RESUME;
			m_veThreadInfos[i]->t_opt = pthread_self();
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
int XJThreadManager::resume_fun(XJ_THR_FUNC func){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->fun == func){
			m_veThreadInfos[i]->opt = XJ_THR_O_RESUME;
			m_veThreadInfos[i]->t_opt = pthread_self();
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
/*方法：线程阻塞，等待子线程退出：
 * */
int XJThreadManager::wait_all(size_t milli){
	int nRet = 0;
	bool bExit = true;
	size_t t = 0;
	while(true){
		pthread_mutex_lock(&m_ThreadMutex);
		std::vector< XJThreadInfo *>::iterator it = m_veThreadInfos.begin();
		while(it != m_veThreadInfos.end()){
			if ((*it)->state != XJ_THR_S_EXITED){
				bExit = false;
			}else if ((*it)->state == XJ_THR_S_EXITED){
				if ((*it)->join){
					void *pRet = NULL;
					pthread_join((*it)->t_id, &pRet);
				}
				delete (*it);
				(*it) = NULL;
				it = m_veThreadInfos.erase(it);
				continue;
			}
			it++;
		}
		pthread_mutex_unlock(&m_ThreadMutex);
		if (bExit){
			break;
		}
		if (t > milli){
			nRet = -1;
			break;
		}
		::usleep(10000);
		t++;
	}
	return nRet;
}
int XJThreadManager::wait(size_t milli, pthread_t t_id){
	int nRet = 0;
	bool bExit = true;
	size_t t = 0;
	while(true){
		pthread_mutex_lock(&m_ThreadMutex);
		std::vector< XJThreadInfo *>::iterator it = m_veThreadInfos.begin();
		while(it != m_veThreadInfos.end()){
			if ((*it)->t_id == t_id && (*it)->state != XJ_THR_S_EXITED){
				bExit = false;
			}else if ((*it)->state == XJ_THR_S_EXITED){
				if ((*it)->join){
					void *pRet = NULL;
					pthread_join((*it)->t_id, &pRet);
				}
				delete (*it);
				(*it) = NULL;
				it = m_veThreadInfos.erase(it);
				continue;
			}
			it++;
		}
		pthread_mutex_unlock(&m_ThreadMutex);
		if (bExit){
			break;
		}
		if (t > milli){
			nRet = -1;
			break;
		}
		::usleep(10000);
		t++;
	}
	return nRet;
}
int XJThreadManager::wait_group(size_t milli, int group_id){
	int nRet = 0;
	bool bExit = true;
	size_t t = 0;
	while(true){
		pthread_mutex_lock(&m_ThreadMutex);
		std::vector< XJThreadInfo *>::iterator it = m_veThreadInfos.begin();
		while(it != m_veThreadInfos.end()){
			if ((*it)->group == group_id && (*it)->state != XJ_THR_S_EXITED){
				bExit = false;
			}else if ((*it)->state == XJ_THR_S_EXITED){
				if ((*it)->join){
					void *pRet = NULL;
					pthread_join((*it)->t_id, &pRet);
				}
				delete (*it);
				(*it) = NULL;
				it = m_veThreadInfos.erase(it);
				continue;
			}
			it++;
		}
		pthread_mutex_unlock(&m_ThreadMutex);
		if (bExit){
			break;
		}
		if (t > milli){
			nRet = -1;
			break;
		}
		::usleep(10000);
		t++;
	}
	return nRet;
}
int XJThreadManager::wait_fun(size_t milli, XJ_THR_FUNC func){
	int nRet = 0;
	bool bExit = true;
	size_t t = 0;
	while(true){
		pthread_mutex_lock(&m_ThreadMutex);
		std::vector< XJThreadInfo *>::iterator it = m_veThreadInfos.begin();
		while(it != m_veThreadInfos.end()){
			if ((*it)->fun == func && (*it)->state != XJ_THR_S_EXITED){
				bExit = false;
			}else if ((*it)->state == XJ_THR_S_EXITED){
				if ((*it)->join){
					void *pRet = NULL;
					pthread_join((*it)->t_id, &pRet);
				}
				delete (*it);
				(*it) = NULL;
				it = m_veThreadInfos.erase(it);
				continue;
			}
			it++;
		}
		pthread_mutex_unlock(&m_ThreadMutex);
		if (bExit){
			break;
		}
		if (t > milli){
			nRet = -1;
			break;
		}
		::usleep(10000);
		t++;
	}
	return nRet;
}
/*方法：获取线程状态
 * */
int XJThreadManager::get_state(pthread_t t_id){
	int nRet = 0;
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == t_id){
			nRet = m_veThreadInfos[i]->state;
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
/*方法：判断线程状态
 * */
bool XJThreadManager::is_state_all(int state){
	bool bRet = true;
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->state != state){
			bRet = false;
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return bRet;
}
bool XJThreadManager::is_state(pthread_t t_id, int state){
	bool bRet = true;
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == t_id &&
				m_veThreadInfos[i]->state != state){
			bRet = false;
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return bRet;
}
bool XJThreadManager::is_state_group(int group_id, int state){
	bool bRet = true;
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->group == group_id &&
				m_veThreadInfos[i]->state != state){
			bRet = false;
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return bRet;
}
bool XJThreadManager::is_state_fun(XJ_THR_FUNC func, int state){
	bool bRet = true;
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->fun == func &&
				m_veThreadInfos[i]->state != state){
			bRet = false;
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return bRet;
}
int XJThreadManager::kill_all(){
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		m_veThreadInfos[i]->opt = XJ_THR_O_EXIT;
		m_veThreadInfos[i]->t_opt = pthread_self();
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return 0;
}
/*方法：线程内部数据获取
 * */
void *XJThreadManager::get_data_self(){
	void *pRet = NULL;
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == pthread_self()){
			pRet = m_veThreadInfos[i]->pData;
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return pRet;
}
/*方法：线程内部数据设置
* */
void *XJThreadManager::set_data_self(void *pData){
	void *pRet = NULL;
	pthread_mutex_lock(&m_ThreadMutex);
	for(size_t i=0;i<m_veThreadInfos.size();i++){
		if (m_veThreadInfos[i]->t_id == pthread_self()){
			pRet = m_veThreadInfos[i]->pData = pData;
			break;
		}
	}
	pthread_mutex_unlock(&m_ThreadMutex);
	return pRet;
}

int XJThreadManager::spawn_i(size_t n,
		XJ_THR_FUNC func,
		void *arg,
		long flags,
		pthread_t *t_id,
		long priority,
		int grp_id,
		size_t stack_size){
	int ret = 0;
	for(int i=0;i<n;i++){
		pthread_mutex_lock(&m_ThreadMutex);
		XJThreadInfo *pThreadInfo = new XJThreadInfo;
		memset(pThreadInfo, 0, sizeof(XJThreadInfo));
		pThreadInfo->group = grp_id;
		pThreadInfo->join = 1;
		pThreadInfo->fun = func;
		pThreadInfo->arg = arg;
		pThreadInfo->exit = 1;
		pThreadInfo->parent = this;
		if (flags & XJ_THR_A_SUSPENDED == XJ_THR_A_SUSPENDED){
			pThreadInfo->opt = XJ_THR_O_SUSPEND;
		}
		if (flags & XJ_THR_A_EXIT_DISABLE == XJ_THR_A_EXIT_DISABLE){
			pThreadInfo->exit = 0;
		}
		pthread_attr_t attrThread;
		pthread_attr_init(&attrThread);
		if (flags & XJ_THR_A_DETACHED == XJ_THR_A_DETACHED){
			pthread_attr_setdetachstate(&attrThread,  PTHREAD_CREATE_DETACHED);
			pThreadInfo->join = 0;
		}
		if (stack_size > 0){
			pthread_attr_setstacksize(&attrThread,  stack_size);
		}

		ret = pthread_create(&pThreadInfo->t_id, &attrThread,
				(void* (*)(void*))XJThreadManager::XJ_Thread_Fun, (void*)pThreadInfo);
		if (ret != 0){
			pthread_mutex_unlock(&m_ThreadMutex);
			return -1;
		}
	}
	return 0;
}
void *XJThreadManager::XJ_Thread_Fun(void *pData){
	XJThreadInfo *pThreadInfo = (XJThreadInfo *)pData;
	pThreadInfo->state = XJ_THR_S_RUNNING;
	pThreadInfo->parent->m_veThreadInfos.push_back(pThreadInfo);
	pthread_mutex_unlock(&m_ThreadMutex);
	int ret = 0;
	ret = pThreadInfo->fun(R_S_BEGIN, pThreadInfo->arg);
	if (ret != 0) {
		return NULL;
	}
	while(true){
		if (pThreadInfo->opt == XJ_THR_O_SUSPEND){
			pthread_mutex_lock(&m_ThreadMutex);
			if (pThreadInfo->opt == XJ_THR_O_SUSPEND){
				pThreadInfo->opt = 0;
				pThreadInfo->state = XJ_THR_S_SUSPENDED;
			}
			pthread_mutex_unlock(&m_ThreadMutex);
			continue;
		}else if (pThreadInfo->opt == XJ_THR_O_RESUME){
			pthread_mutex_lock(&m_ThreadMutex);
			if (pThreadInfo->opt == XJ_THR_O_RESUME){
				pThreadInfo->opt = 0;
				pThreadInfo->state = XJ_THR_S_RUNNING;
			}
			pthread_mutex_unlock(&m_ThreadMutex);
			break;
		}else if (pThreadInfo->opt == XJ_THR_O_EXIT){
			pthread_mutex_lock(&m_ThreadMutex);
			if (pThreadInfo->opt == XJ_THR_O_EXIT) {
				pThreadInfo->opt = 0;
			}
			pthread_mutex_unlock(&m_ThreadMutex);
			break;
		}
		if (pThreadInfo->state == XJ_THR_S_SUSPENDED){
			::usleep(10000);
			continue;
		}
		ret = pThreadInfo->fun(R_S_RUNING, pThreadInfo->arg);
		if (ret!=0){
			break;
		}
	}
	pThreadInfo->fun(R_S_END, pThreadInfo->arg);
	pthread_mutex_lock(&m_ThreadMutex);
	pThreadInfo->state = XJ_THR_S_EXITED;
	pthread_mutex_unlock(&m_ThreadMutex);
	return NULL;
}

} /* namespace XJUtilClass */
